package com.lihh.rocket5.transaction;

import com.lihh.rocket5.transaction.impl.TransactionListenerImpl;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TransactionProducer {
  public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
    // 1. 创建 事务监听器
    TransactionListener transactionListener = new TransactionListenerImpl();
    // 2. 创建 发送事务的 producer
    TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-group");
    // 3. 设置 name server
    producer.setNamesrvAddr("localhost:9876");
    // 4. 基于 JUC 创建线程池
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new ThreadFactory() {
      @Override
      public Thread newThread(Runnable r) {
        Thread thread = new Thread(r);
        thread.setName("transaction producer ...");
        return thread;
      }
    });
    
    // 5. 设置生产者回查线程
    producer.setExecutorService(threadPoolExecutor);
    // 6. 设置生产者监听器
    producer.setTransactionListener(transactionListener);
    // 7. 开始启动生产者
    producer.start();
    
    // 发送 事务消息
    Message message = new Message("TransactionTopic", null, ("用户A 向 用户B 转账100元").getBytes(RemotingHelper.DEFAULT_CHARSET));
    TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);
    
    // 判断发送是否成功
    if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
      SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
      System.out.println("1. 发送半事务消息成功:" + df.format(new Date()));//半事务消息是否成功
    } else {
      System.out.println("发送半事务消息失败.");
    }
    
    while (true) {}
//    producer.shutdown();
  }
}
